package cn.me.alphamq.common.entity;

import cn.me.alphamq.common.constant.GlobalConstant;
import cn.me.alphamq.common.entity.req.PushDelayMsgReq;
import cn.me.alphamq.common.entity.req.PushMsgReq;
import cn.me.alphamq.common.entity.req.TxCallbackCheckReq;

/**
 * 业务的消息对象
 *
 * @author f
 */
public class Message {

    /**
     * 消息的主题，由 生产者 指定
     */
    private String topic;

    /**
     * 放入消息的队列 id，默认值是 -1，如果没有传递，则会在写入数据时默认写到 0 号队列，由 生产者/broker 指定
     */
    private int queueId = GlobalConstant.DEFAULT_QUEUE_ID;

    /**
     * 消息的主体，由 生产者/broker 指定
     */
    private byte[] body;

    /**
     * 消息存放的偏移量，由 消费者 指定
     */
    private int msgOffset;

    /**
     * 消息的长度，由 消费者 指定
     */
    private int msgLength;

    /**
     * 当前重试次数，由 broker 指定
     */
    private int curRetryTimes;

    /**
     * 简化消息，让消息只包含 偏移量 和 长度
     *
     * @param message 被简化的消息
     * @return 简化后的消息
     */
    public static Message simplify(Message message) {
        message.msgLength = message.body.length;
        message.body = null;
        return message;
    }

    public Message() {
    }

    public Message(String retryTopic, int queueId, byte[] commitLogBody, int curRetryTimes) {
        this.topic = retryTopic;
        this.queueId = queueId;
        this.body = commitLogBody;
        this.curRetryTimes = curRetryTimes;
    }

    /**
     * 生产者 使用的构造器，用于构建一条指定 topic 和 内容 的消息
     *
     * @param topic topic
     * @param body  内容
     */
    public Message(String topic, byte[] body) {
        this.topic = topic;
        this.body = body;
    }

    /**
     * 生产者 使用的构造器，用于构建一条指定 topic、队列 id 和 内容 的消息
     *
     * @param topic   topic
     * @param queueId 队列 id
     * @param body    内容
     */
    public Message(String topic, int queueId, byte[] body) {
        this.topic = topic;
        this.queueId = queueId;
        this.body = body;
    }

    /**
     * broker 使用的构造器，用于构建一条指定 内容、偏移量 和 当前重试次数 的消息
     *
     * @param body          内容
     * @param msgOffset     偏移量
     * @param curRetryTimes 当前重试次数
     */
    public Message(byte[] body, int msgOffset, int curRetryTimes) {
        this.body = body;
        this.msgOffset = msgOffset;
        this.curRetryTimes = curRetryTimes;
    }

    /**
     * broker 使用的构造器，用于从 生产者生产的消息 中构建一条指定 内容 和 偏移量 的消息
     *
     * @param pushMsgReq 指定的消息
     */
    public Message(PushMsgReq pushMsgReq) {
        this(pushMsgReq.getTopic(), pushMsgReq.getQueueId(), pushMsgReq.getBody());
    }

    /**
     * broker 使用的构造器，用于从 生产者生产的消息 中构建一条指定 内容 和 偏移量 的消息
     *
     * @param pushDelayMsgReq 指定的延迟消息
     */
    public Message(PushDelayMsgReq pushDelayMsgReq) {
        this(pushDelayMsgReq.getTopic(), pushDelayMsgReq.getQueueId(), pushDelayMsgReq.getBody());
    }

    /**
     * broker 使用的构造器，用于事务消息写入 commitLog 文件中
     *
     * @param topic         事务消息的主题
     * @param queueId       事务消息存放的队列 id
     * @param commitLogBody 原消息的主体
     */
    public static Message newTxMsg(String topic, int queueId, byte[] commitLogBody) {
        return new Message(topic, queueId, commitLogBody);
    }

    /**
     * broker 使用的构造器，用于将延迟消息写入延迟队列中
     * <p>
     * 默认延迟消息只有 0 号队列
     *
     * @param delayTopic    延迟队列的主题
     * @param queueId       写入延迟队列的队列 id
     * @param commitLogBody 原消息的主体
     */
    public static Message newDelayMsg(String delayTopic, int queueId, byte[] commitLogBody) {
        return new Message(delayTopic, queueId, commitLogBody);
    }

    /**
     * broker 使用的构造器，用于消费失败后将原消息写入重试队列中
     *
     * @param retryTopic    重试队列的主题
     * @param queueId       写入重试队列的队列 id
     * @param commitLogBody 原消息的主体
     * @param curRetryTimes 当前重试次数
     */
    public static Message newRetryMsg(String retryTopic, int queueId, byte[] commitLogBody, int curRetryTimes) {
        return new Message(retryTopic, queueId, commitLogBody, curRetryTimes);
    }

    /**
     * 客户端 使用的构造器，用于检查本地事务的状态
     *
     * @param txCallbackCheckReq 检查本地事务状态的请求对象
     */
    public static Message newCheckTxStateMsg(TxCallbackCheckReq txCallbackCheckReq) {
        return new Message(txCallbackCheckReq.getTopic(), txCallbackCheckReq.getQueueId(), txCallbackCheckReq.getBody());
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public int getQueueId() {
        return queueId;
    }

    public void setQueueId(int queueId) {
        this.queueId = queueId;
    }

    public byte[] getBody() {
        return body;
    }

    public void setBody(byte[] body) {
        this.body = body;
    }

    public int getMsgOffset() {
        return msgOffset;
    }

    public void setMsgOffset(int msgOffset) {
        this.msgOffset = msgOffset;
    }

    public int getMsgLength() {
        return msgLength;
    }

    public void setMsgLength(int msgLength) {
        this.msgLength = msgLength;
    }

    public int getCurRetryTimes() {
        return curRetryTimes;
    }

    public void setCurRetryTimes(int curRetryTimes) {
        this.curRetryTimes = curRetryTimes;
    }
}
